package com.cbh.amadeus.server.handler;

import com.cbh.amadeus.common.enums.MessageType;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.protocol.AmadeusMessageProtocol;
import com.cbh.amadeus.common.protocol.AmadeusMessageWrap;
import com.cbh.amadeus.server.core.Publisher;
import com.cbh.amadeus.server.domain.AckResultModel;
import com.cbh.amadeus.server.domain.ChannelModel;
import com.cbh.amadeus.server.provider.component.PublisherComponent;
import com.cbh.amadeus.server.provider.tool.AckResultModelTool;
import com.cbh.amadeus.server.utils.MessageRejectManage;
import com.cbh.amadeus.server.utils.TopicChannelManage;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author ChenBiHui
 * @date 2025/5/1
 */
@Slf4j
public class ResendHandler extends SimpleChannelInboundHandler<AmadeusMessageWrap> {

    private final PublisherComponent component;

    public ResendHandler(PublisherComponent component) {
        this.component = component;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, AmadeusMessageWrap msg) {
        MessageType type = MessageType.find(msg.getType());
        if (type == MessageType.RESEND) {
            AmadeusMessage message = msg.getMessage();
            String topic = message.getTopic();
            String messageId = message.getMessageId();
            List<ChannelModel> models = TopicChannelManage.findModelWithTopic(topic);
            if (models == null || models.isEmpty()) {
                ctx.writeAndFlush(MessageRejectManage.reject(msg))
                        .addListener(item -> {
                            if (item.isSuccess()) {
                                log.warn(
                                        "===== [RESEND REJECT] Topic [{}] message-id [{}] can't find any channel, has reject this message =====",
                                        topic,
                                        messageId
                                );
                            }
                        });
                return;
            }
            Publisher.DEFAULT.execute(() -> {
                AckResultModelTool tool = component.getModelTool();
                Set<String> ack = tool.match(topic, messageId)
                        .stream()
                        .map(AckResultModel::getConsumer)
                        .collect(Collectors.toSet());
                List<ChannelModel> unAck = models.stream()
                        .filter(item -> !ack.contains(item.getOrigin()))
                        .collect(Collectors.toList());
                if (unAck.isEmpty()) {
                    ctx.writeAndFlush(release(message))
                            .addListener(item -> {
                                if (item.isSuccess()) {
                                    log.info(
                                            "===== [RESEND] Topic [{}] message-id [{}] has back to release =====",
                                            topic,
                                            messageId
                                    );
                                }
                            });
                    return;
                }
                for (ChannelModel model : unAck) {
                    Channel channel = model.getChannel();
                    if (channel.isActive()) {
                        channel.writeAndFlush(msg)
                                .addListener(item -> {
                                    if (item.isSuccess()) {
                                        log.info(
                                                "===== [RESEND] Topic [{}] message-id [{}] has resend to the register [{}] =====",
                                                message.getTopic(),
                                                message.getMessageId(),
                                                model.getOrigin()
                                        );
                                    }
                                });
                    }
                }
            });
            return;
        }
        ctx.fireChannelRead(msg);
    }

    private AmadeusMessageWrap release(AmadeusMessage message) {
        AmadeusMessageWrap wrap = new AmadeusMessageWrap();
        wrap.setMagic(AmadeusMessageProtocol.MAGIC);
        wrap.setType(MessageType.RELEASE.getType().byteValue());
        wrap.setMessage(
                new AmadeusMessage()
                        .setProvider(message.getProvider())
                        .setTopic(message.getTopic())
                        .setMessageId(message.getMessageId())
        );
        return wrap;
    }
}
